package com.kongxiang.storm.bolt;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.storm.jdbc.bolt.JdbcInsertBolt;
import org.apache.storm.jdbc.common.Column;
import org.apache.storm.jdbc.common.ConnectionProvider;
import org.apache.storm.jdbc.common.HikariCPConnectionProvider;
import org.apache.storm.jdbc.mapper.JdbcMapper;
import org.apache.storm.jdbc.mapper.SimpleJdbcMapper;


import java.sql.Types;
import java.util.List;
import java.util.Map;

/**
 * @version 1.0
 * @description:
 * @projectName: com.kongxiang.storm.bolt
 * @className: demo
 * @author:谭农春
 * @createTime:2018/9/29 21:26
 */
public class JDBCBolt {


  public  static final JdbcInsertBolt getJdbcInsertBolt(final String tableName){
      // 配置数据
      Map<String,Object> map = Maps.newHashMap();
      map.put("dataSourceClassName", "com.mysql.jdbc.jdbc2.optional.MysqlDataSource");
      map.put("dataSource.url", "jdbc:mysql://localhost:3306/logfx");
      map.put("dataSource.user", "root");//root
      map.put("dataSource.password", "password");
      ConnectionProvider connectionProvider = new HikariCPConnectionProvider(map);
      connectionProvider.prepare();
    // 插入到数据库
    //must specify column schema when providing custom query.
  //  List<Column> schemaColumns = Lists.newArrayList(new Column("message", Types.VARCHAR));
//    JdbcMapper mapper = new SimpleJdbcMapper(schemaColumns);
    JdbcMapper mapper = new SimpleJdbcMapper(tableName,connectionProvider);
    JdbcInsertBolt persistanceBolt = new JdbcInsertBolt(connectionProvider, mapper);
    persistanceBolt.withTableName(tableName);
    //persistanceBolt.withInsertQuery("insert into other_log (message) values (?) ");
    return  persistanceBolt;
  }
}
